"""
Case Type   : 操作系统类-文件系统故障
Case Name   : 逻辑复制时 主机注入故障系统文件句柄耗尽
Create At   : 2021/8/3
Owner       : li-xin12345
Description :
    1、逻辑复制准备
    2、注入文件系统inode资源耗尽故障
    3、启线程执行创建逻辑复制槽
    4、清理故障
    5、再次执行创建逻辑复制槽
    6、删除逻辑复制槽
    7、运行TPCC
    8、检查数据一致性
    9、恢复环境
Expect      :
    1、逻辑复制准备成功
    2、注入故障成功
    3、创建逻辑复制槽失败
    4、清理故障成功
    5、创建逻辑复制槽成功
    5、gsql恢复报错“Too many open files”
    6、删除逻辑复制槽成功
    7、运行TPCC成功
    8、数据一致
    9、恢复环境
History     :
"""
import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.CommonSH import RestartDbCluster
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node, macro


class ReliabilityFileSystem(unittest.TestCase):
    nodes = ("PrimaryDbUser", "Standby1DbUser", "Standby2DbUser")

    @RestartDbCluster(*nodes)
    def setUp(self):
        self.log = Logger()
        self.primary_node = Node(self.nodes[0])
        self.primary_sh = CommonSH(self.nodes[0])
        self.primary_root_node = Node(node="PrimaryRoot")
        self.constant = Constant()
        self.com = Common()
        self.file = os.path.join(macro.DB_INSTANCE_PATH,
                                 macro.DB_PG_CONFIG_NAME)
        self.log.info("setUp-实例化:ok")

        self.log.info("Opengauss_Reliability_OS_File_System_Case060:开始执行")
        self.inject_cmd = f"rfile_deplenish (timeout) values (300)"
        self.pg_hba = os.path.join(macro.DB_INSTANCE_PATH,
                                   macro.PG_HBA_FILE_NAME)

        self.log.info("备份pg_hba文件")
        shell = f"cp {self.pg_hba} {macro.DB_INSTANCE_PATH}/bk_hba.conf"
        result = self.com.get_sh_result(self.primary_node, shell)
        self.log.info(result)
        self.assertNotIn("bash", result)

    def test_main(self):
        text = "--step1：逻辑复制准备"
        self.log.info(text)
        self.log.info("修改wal_level为logical;enable_slot_log为on")
        result = self.primary_sh.execute_gsguc("set",
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               "wal_level =logical")
        self.log.info(result)
        self.assertTrue(result, "执行失败" + text)
        result = self.primary_sh.execute_gsguc("set",
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               "enable_slot_log =on")
        self.log.info(result)
        self.assertTrue(result, "执行失败" + text)

        self.log.info("重启数据库")
        restart_msg = self.primary_sh.restart_db_cluster()
        self.log.info(restart_msg)
        status = self.primary_sh.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        self.log.info("主机配置逻辑复制的用户")
        sql_cmd = self.primary_sh.executDbSql(
            f"drop role if exists rep;"
            f"create role test_rep with login password "
            f"'{macro.COMMON_PASSWD}';"
            f"alter role test_rep with replication sysadmin;")
        self.log.info(sql_cmd)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, sql_cmd)
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, sql_cmd)

        mod_msg = f"sed -i '$a\local    replication     test_rep1     " \
                  f"trust' {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host    replication     test_rep1  " \
                  f"127.0.0.1/32   trust' {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)
        mod_msg = f"sed -i '$a\host   replication     test_rep1  " \
                  f"::1/128    trust'  {self.pg_hba}"
        self.log.info(mod_msg)
        msg = self.primary_node.sh(mod_msg).result()
        self.log.info(msg)

        try:
            text = "--step2:主节点注入句柄耗尽故障;expect:成功"
            self.log.info(text)
            self.log.info("注入故障")
            session1 = ComThread(self.com.cfe_inject,
                                 args=(self.primary_root_node,
                                       self.inject_cmd))
            session1.setDaemon(True)
            session1.start()
            time.sleep(1)

            text = "--step3：执行创建逻辑复制槽"
            self.log.info(text)
            self.log.info("主机创建逻辑复制槽")
            create_cmd = f"pg_recvlogical -d postgres " \
                         f"-U test_rep" \
                         f"-S slot_test001 " \
                         f"-p {self.primary_node.db_port} " \
                         f"-P mppdb_decoding " \
                         f"--create"
            execute_cmd = f'''source {macro.DB_ENV_PATH}
                       expect <<EOF
                       set timeout 300
                       spawn {create_cmd}
                       expect "Password:"
                       send "{macro.COMMON_PASSWD}\\n"
                       expect eof\n''' + '''EOF'''
            self.log.info(execute_cmd)
            session2 = ComThread(self.com.get_sh_result,
                                 args=(self.primary_node,
                                       execute_cmd))
            session2.setDaemon(True)
            session2.start()

            self.log.info("创建逻辑复制槽结果")
            session2.join(60 * 5)
            result = session2.get_result()
            self.log.info(result)

            self.assertIn(self.constant.TOO_MANYY_FILE_MSG, result,
                          "执行失败" + text)

            self.log.info("故障注入结果")
            session1.join(60 * 5)
            result = session1.get_result()
            self.log.info(result)
            self.assertIn("successful execution", result, "执行失败" + text)

        finally:
            text = "--step4:清除故障;expect:成功"
            self.log.info(text)
            for i in range(5):
                result = self.com.cfe_query(self.primary_root_node,
                                            "rfile_deplenish")
                self.log.info(result)
                if self.constant.TOO_MANYY_FILE_MSG in result:
                    time.sleep(60)
                else:
                    break
            result = self.com.cfe_query(self.primary_root_node,
                                        "rfile_deplenish")
            self.log.info(result)
            self.assertNotIn(self.constant.TOO_MANYY_FILE_MSG, result,
                             "执行失败" + text)

            text = "--step5:再次创建指定实例的备份"
            self.log.info(text)
            create_cmd = f"pg_recvlogical -d postgres " \
                         f"-U test_rep" \
                         f"-S slot_test001 " \
                         f"-p {self.primary_node.db_port} " \
                         f"-P mppdb_decoding " \
                         f"--create"
            execute_cmd = f'''source {macro.DB_ENV_PATH}
                       expect <<EOF
                       set timeout 300
                       spawn {create_cmd}
                       expect "Password:"
                       send "{macro.COMMON_PASSWD}\\n"
                       expect eof\n''' + '''EOF'''
            self.log.info(execute_cmd)
            exec_msg = self.primary_node.sh(execute_cmd).result()
            self.log.info(exec_msg)
            self.assertNotIn(self.constant.TOO_MANYY_FILE_MSG, exec_msg)
            self.assertIn("total time", exec_msg)

            query_cmd = self.primary_sh.executDbSql(
                "select slot_name,plugin from"
                " pg_get_replication_slots();")
            self.log.info(query_cmd)
            self.assertIn("slot_test001", query_cmd)

            text = "--step6：主机删除复制槽"
            self.log.info(text)
            del_cmd = f"pg_recvlogical -d postgres " \
                      f"-U test_rep" \
                      f"-S slot_test001 " \
                      f"-p {self.primary_node.db_port} " \
                      f" --drop"
            self.log.info(del_cmd)
            execute_cmd = f'''source {macro.DB_ENV_PATH}
                              expect <<EOF
                              set timeout 300
                              spawn {del_cmd}
                              expect "Password:"
                              send "{macro.COMMON_PASSWD}\\n"
                              expect eof\n''' + '''EOF'''
            self.log.info(execute_cmd)
            exec_msg = self.primary_node.sh(execute_cmd).result()
            self.log.info(exec_msg)
            self.assertNotIn(self.constant.SQL_WRONG_MSG[1], exec_msg)

        db_status = self.primary_sh.getDbClusterStatus("detail")
        self.log.info(db_status)
        self.assertTrue("Degraded" in db_status or "Normal" in db_status)

        text = "--step7:运行TPCC;expect:成功"
        self.log.info(text)
        result = self.com.startTPCC(self.primary_root_node, macro.TPCC_PATH)
        self.log.info(result)
        self.assertIn(self.constant.TPCC_SUCCESS_MSG, result, "执行失败" + text)

        self.log.info("检查主备是否同步")
        time.sleep(10)
        self.log.info("获取节点数量")
        node_num = self.primary_sh.get_node_num()
        res = self.primary_sh.check_location_consistency("primary", node_num)
        self.assertTrue(res, "执行失败" + text)

        text = "--step8:检查数据一致性"
        self.log.info(text)
        nodes = (Node(self.nodes[0]), Node(self.nodes[1]),
                 Node(self.nodes[2]))
        flag = self.com.check_data_sample_by_all(r'\d', *nodes)
        self.assertTrue(flag, "执行失败" + text)
        table_dict = self.com.format_sql_result(
            self.primary_sh.executDbSql(r'\d'))
        table_name = table_dict.get('Name')
        for name in table_name:
            select_sql = f"select count(*) from {name};"
            flag = self.com.check_data_sample_by_all(select_sql, *nodes)
            self.assertTrue(flag, "执行失败" + text)

    def tearDown(self):
        text = "--step9:环境清理;expect:成功"
        self.log.info(text)
        result = self.com.cfe_query(self.primary_root_node,
                                    "rfile_deplenish")
        self.log.info(result)
        if self.constant.TOO_MANYY_FILE_MSG in result:
            time.sleep(300)

        self.log.info("清理逻辑复制槽相关信息")
        del_cmd = f"pg_recvlogical -d postgres " \
                  f"-U test_rep" \
                  f"-S slot_test001 " \
                  f"-p {self.primary_node.db_port} " \
                  f" --drop"
        self.log.info(del_cmd)
        execute_cmd = f'''source {macro.DB_ENV_PATH}
                          expect <<EOF
                          set timeout 300
                          spawn {del_cmd}
                          expect "Password:"
                          send "{macro.COMMON_PASSWD}\\n"
                          expect eof\n''' + '''EOF'''
        self.log.info(execute_cmd)
        exec_msg = self.primary_node.sh(execute_cmd).result()
        self.log.info(exec_msg)
        sql_cmd = self.primary_sh.executDbSql("drop role if exists rep;")
        self.log.info(sql_cmd)
        result = self.primary_sh.execute_gsguc("set",
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               "wal_level=hot_standby")
        self.log.info(result)
        result = self.primary_sh.execute_gsguc("set",
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               "enable_slot_log=off")
        self.log.info(result)

        self.log.info("恢复pg_hba文件")
        shell = f"ls {macro.DB_INSTANCE_PATH}/bk_hba.conf"
        self.log.info(shell)
        result = self.com.get_sh_result(self.primary_node, shell)
        if "No such file or directory" not in result:
            shell = f"mv {macro.DB_INSTANCE_PATH}/bk_hba.conf {self.pg_hba}"
            self.log.info(shell)
            result = self.com.get_sh_result(self.primary_node, shell)
            self.log.info(result)
            self.assertNotIn("bash", result, "执行失败" + text)

        result = Node(self.nodes[0]).sh("date").result()
        self.assertNotIn(self.constant.TOO_MANYY_FILE_MSG, result,
                         "执行失败" + text)
        self.log.info("故障已清理")
        self.log.info("Opengauss_Reliability_OS_File_System_Case060:执行完成")
